## SpringRetry重试机制
![image-20230623113044603](https://img.wkq.pub/img/image-20230623113044603.png)
![image-20230623113132888](https://img.wkq.pub/img/image-20230623113132888.png)

## Spring异步任务
## RabbitMQ异步解耦
### 什么是 MQ ？
MQ 的全称是 Message Queue，即消息队列。消息队列是在消息传输的过程中，会把消息保存到一个容器里面 ，这个容器就是队列，消息就是生产出来的一些信息数据。主要是一些字符串类型的，json 字符串，普通的字符串，甚至是数值的字符串，都可以存储在消息队列中，消息队列主要是应用在一些分布式或微服务系统中，用来解耦合，异步处理，削峰填谷等等。

系统通信，可以直接通过  REST 方式调用，也可以通过一个中间媒介进行调用，中间媒介负责解耦/

![image-20230623180015132](https://img.wkq.pub/img/image-20230623180015132.png)

### MQ 选型

* RabbitMQ： RabbitMQ 综合能力比较高，足够满足自身业务。
* ActiveMQ
* RocketMQ：吞吐量非常高，高并发能力远比RabbitMQ更高，功能非常全面，扩展性也很高，金融类项目优先使用 RocketMQ 。
* Kafka： 在大数据类型应用比较多，吞吐量也要比 RabbitMQ 更高。

### RabbitMQ

* Erlang，内部开发语言。
* AMQP协议（Advanced Message Queuing Protocol）。应用消息中间件的一个协议。
* https://www.rabbitmq.com

### 应用场景

* 异步任务：最终一致性
* 提速，减少等待时间，提高用户体验
* 解耦、广播机制（比如下单的同时要调用库存，支付、物流、日志、积分等等多个服务，如果说是同步进行的，一旦某一个环节出了问题，整个接口就直接崩掉了，如果直接使用 MQ 进行解耦，那么哪怕要换也是只换一个环节，因此就大大减小了发生问题的风险）
* 削峰填谷，高并发的一个场景，针对系统来说就非常稳定了。

​    ![image-20230623183946632](https://img.wkq.pub/img/image-20230623183946632.png)

### RabbitMQ 模型

Producer 和 Consumer 都是客户端，Channel 是一个连接的通道，相当于 HTTP Session 的一个会话，建立请求就会有 Channel（桥梁）。消息必须要经过桥梁才能到达。

Channel 的作用就是为了更好的管理每次请求到 MQ 的连接（Socket 连接），如此才能更好的节约开支，直接去提高资源的效率。

Exchange 是交换机，是消息队列的交换机，它会按照一定的规则来转发消息到某一个队列（分发），就相当于 Controller 的路由机制。

队列是用于存储消息的，可以把它理解为是数据库里面的某张表，每张表都可以去存储一些数据，队列是用于去存储生产者发过来的消息。消费者是基于监听的，得到消息之后就可以从队列里获得相应的消息了。消息存储是有一定的机制的，

Exchange 和 Queue 是有绑定的关系，绑定的关系其实也就是 路由的关系，交换器和队列存在绑定关系是和路由结合在一起的。通过路由机制，消息从交换机发送到队列，路由机制就相当于 Controller 里面所使用的 RequestMapping，路由规则一旦匹配，就可以把消息存储到队列里了。

![image-20230623191212648](https://img.wkq.pub/img/image-20230623191212648.png)

### Docker 安装配置 RabbitMQ

```shell
docker run -d -p 5671:5671 -p 5672:5672 -p 15671:15671 -p 15672:15672 -p 25672:25672 --restart always --name rabbitmq rabbitmq:management
```

## RabbitMQ工作模式

###  简单模式





![Producer -> Queue -> Consuming: send and receive messages from a named queue.](https://img.wkq.pub/img/python-one.png)

### 工作队列

![Producer -> Queue -> Consuming: Work Queue used to distribute time-consuming tasks among multiple workers.](https://img.wkq.pub/img/python-two.png)

在原先的简单模式变为了两个消费者共同监听队列。工作队列也没有交换机，队列中的消息由多个消费者竞争消费。

队列中的消息只会被某一个消费者进行消费（消费者负载均衡）。

### 发布订阅模式

![Producer -> Queue -> Consuming: deliver a message to multiple consumers. This pattern is known as publish/subscribe](https://img.wkq.pub/img/python-three.png)

生产者和队列之间增加了一个 X，X 就是一个 Exchange，工作队列是一个消息只能被一个消费者去消费。

发布订阅模式则不同，只要监听，所有消费者都能够消费同一个消息。

不同的用户群都可以接收到相同的消息了。

队列就相当于公众号，通过交换机把消息路由到不同的队列，就可以让不同的用户群接收到不同的消息了。

发布和订阅是针对交换机的。

### 路由模式

![Producer -> Queue -> Consuming: subscribe to a subset of the messages only.](https://img.wkq.pub/img/python-four.png)

路由模式是在发布订阅模式的基础上增加了路由的功能，路由模式是通过路由键来进行消息的路由的。

### 通配符模式

![Producer -> Queue -> Consuming: receiving messages based on a pattern (topics).](https://img.wkq.pub/img/python-five.png)

* `*`表示匹配一个单词
* `#` 表示匹配多个单词

## jasypt 配置文件密码加解密
引入依赖

```xml
<dependency>
    <groupId>com.github.ulisesbocchio</groupId>
    <artifactId>jasypt-spring-boot-starter</artifactId>
    <version>3.0.3</version>
</dependency>
```
加解密机制
```java
@SpringBootTest
public class JasyptTest {
    @Test
    public void testPwdEncrypt() {
        //1. 实例化加密器
        StandardPBEStringEncryptor encryptor = new StandardPBEStringEncryptor(); //默认的加密器
        //2. 配置加密算法和密钥
        EnvironmentPBEConfig config = new EnvironmentPBEConfig();
        config.setAlgorithm("PBEWithMD5AndDES"); //加密算法
        config.setPassword("hire"); //密钥 (类似盐，可以是随机字符串或者固定，一定要存储好)
        encryptor.setConfig(config);
        //3. 对密码进行加密
        String myPassword = "Zbb010719!";
        String encryptedPassword = encryptor.encrypt(myPassword);
        System.out.println(encryptedPassword);

        //4. 对密码进行解密
        String decryptedPassword = encryptor.decrypt(encryptedPassword);
        System.out.println(decryptedPassword);
        // 每次加密后的字符串密码都不一样，但是每次解密过后都是一样的。  cJZvwxKnvyK+ICGbPwkj4J5LQjP+JbMm  nZA5ZRRDWAafsD8PWD9ksCt29EzcRKzI
    }
}
```
配置文件加密
```yaml
spring:
  datasource:
    url: jdbc:mysql://localhost:3306/hire?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai
    username: root
    password: ENC(cJZvwxKnvyK+ICGbPwkj4J5LQjP+JbMm)
    driver-class-name: com.mysql.cj.jdbc.Driver
```
## 消息的可靠性投递

### Configm机制

![image-20230702112308693](https://img.wkq.pub/img/image-20230702112308693.png)

```java
   //定义confirm回调
        rabbitTemplate.setConfirmCallback((/*相关联数据，参数*/correlationData,/*交换机是否成功接收到消息*/ ack,/*失败的原因*/ cause) -> {
            if (ack) {
                log.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause);
            } else {
                log.info("消息发送失败:correlationData({}),ack({}),cause({})", correlationData, ack, cause);
            }
        });
        rabbitTemplate.convertAndSend(SMS_EXCHANGE_NAME, SEND_EMAIL_ROUTING_KEY,
                SerializationUtils.serializeObject(smsContentQO),
                new CorrelationData(UUID.randomUUID().toString()));
```

### Return 机制

```java
  //定义 return 回调
        rabbitTemplate.setReturnsCallback((returnCallback) -> {
            log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message({})",
                    returnCallback.getExchange(), returnCallback.getRoutingKey(), returnCallback.getReplyCode(),
                    returnCallback.getReplyText(), returnCallback.getMessage());
        });
        rabbitTemplate.convertAndSend(SMS_EXCHANGE_NAME, "abc."+SEND_EMAIL_ROUTING_KEY,
                SerializationUtils.serializeObject(smsContentQO),
                new CorrelationData(UUID.randomUUID().toString()));
```

```shell
消息丢失:exchange(sms_exchange),route(abc.sms.mail.send.login),replyCode(312),replyText(NO_ROUTE),message((Body:'{"mobile":"2112954786@qq.com","content":"891093"}' MessageProperties [headers={spring_returned_message_correlation=d2ee6263-3e40-4045-83c2-edf5d6673965}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0]))
```

RabbitMQ 也有事务机制，但是在高并发的情况下性能会比较低，所以不推荐使用

## 消费端 ACK

* Acknowledge
* Acknowledge=none（自动确认，消费者收到消息之后，就会给 Broker 一个确认，不管业务是处理成功还是失败）
* Acknowledge=manual(手动确认，在监听的方法里处理完业务流程，手动进行确认)
* Acknowledge=auto（异常类型确认，了解即可）

```java

    /**
     * 手动确认监听短信消息，通过 channel 进行手动的确认
     *
     * @param message 消息
     */
    @RabbitListener(queues = {RabbitMQSMSConfig.SMS_QUEUE_NAME})
    public void receiveSMSaMessage(Message message, Channel channel) {
        try {
            //获取消息体
            byte[] body = message.getBody();
            //获取消息体的字符串
            String payload = new String(body);
            log.info("接收到消息：payload: {} : message : {}", payload, message);
            String receivedRoutingKey = message.getMessageProperties().getReceivedRoutingKey();
            log.info("接收到消息：routingKey: {}", receivedRoutingKey);

            if (Objects.equals(receivedRoutingKey, RabbitMQSMSConfig.SEND_SMS_ROUTING_KEY)) {
                log.info("接收到消息：发送短信");
            } else if (Objects.equals(receivedRoutingKey, RabbitMQSMSConfig.SEND_EMAIL_ROUTING_KEY)) {
                log.info("接收到消息：发送邮件");
                //此处为邮件发送的消息发送处理逻辑
                SMSContentQO smsContentQO = SerializationUtils.deserializeObject(payload, SMSContentQO.class);
                mailUtils.sendSimpleMail("2112954786@qq.com", smsContentQO.getMobile(), "登录验证码", "您的登录验证码为：" + smsContentQO.getContent() + "，请勿泄露给他人。");
                //手动确认消息
                /**
                 * 参数一：消息的唯一标识
                 * 参数二：是否批量确认，true：批量确认所有小于当前消息唯一标识的消息
                 *       false：只确认当前消息
                 *       如果为true，会将当前消息之前的所有消息都确认
                 */
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }
        } catch (Exception e) {
            log.error("接收消息发生异常：{}", e.getMessage());
            try {
                /**
                 * 参数一：消息的唯一标识
                 * 参数二：是否批量确认，true：批量确认所有小于当前消息唯一标识的消息
                 *      false：只确认当前消息
                 *      如果为true，会将当前消息之前的所有消息都确认
                 * 参数三：是否重回队列，true：重回队列，false：不重回队列,丢弃消息
                 */
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
            } catch (IOException ex) {
                throw new RuntimeException(ex);
            }
        }
    }
```

## 消费者消息限流

## 消息超时

![image-20230702150620561](https://img.wkq.pub/img/image-20230702150620561.png)

让超时消息失效，让 MQ 直接把它抛弃。

TTL 存活时间

time to live

同 redis ttl

到达 ttl 时间节点，消费未消费则抛弃

可以对   队列/单个消息， 设置 ttl

## 死信队列

![image-20230702235557111](https://img.wkq.pub/img/image-20230702235557111.png)

* ttl 超时丢弃
* 超出队列长度限制被丢弃
* 手动 nack 或 reject，并且 requeue 未 false
